package com.sdses.flink.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.File;

public class FlinkSQL_JsonArray {
    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

//        System.setProperty("java.io.tmpdir", new File(MSApp.instance().getAppPaths().getTempDir() , "flink").getAbsolutePath()) ;

        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    order_number BIGINT,\n" +
                "    price        DECIMAL(32,2),\n" +
                "    buyer        ROW<first_name STRING, last_name STRING>,\n" +
                "    order_time   TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'rows-per-second' = '5', \n" +
                "  'number-of-rows' = '10', \n" +
//                "  'fields.order_number.kind' = 'sequence', \n" +
//                "  'fields.order_number.start' = '1', \n" +
//                "  'fields.order_number.end' = '2', \n" +
                "  'fields.order_number.min' = '3', \n" +
                "  'fields.order_number.max' = '5', \n" +

                "  'fields.price.min' = '10', \n" +
                "  'fields.price.max' = '100', \n" +

                "  'fields.buyer.first_name.length' = '5', \n" +
                "  'fields.buyer.last_name.length' = '5' \n" +
                ")");


//        tableEnv.executeSql("CREATE TABLE sinkTable(field string) WITH ('connector' = 'print')");

//        tableEnv.executeSql("CREATE TABLE sinkTable(field1 BIGINT, field2  ROW<first_name STRING, last_name STRING>, PRIMARY KEY (field1) NOT ENFORCED ) WITH (" +
//                "  'connector' = 'upsert-kafka',\n" +
//                "  'topic' = 'test_ods_data',\n" +   //
//                "  'properties.bootstrap.servers' = '192.168.102.120:9092,192.168.102.121:9092,192.168.102.122:9092,192.168.102.123:9092',\n" +
//                "  'key.format' = 'json',\n" +
//                "  'value.format' = 'json'\n" +
//                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable(field1 BIGINT, field2  ROW<first_name STRING, last_name STRING>) WITH (" +
//                "  'connector' = 'kafka',\n" +
//                "  'topic' = 'test_ods_data',\n" +   //
//                "  'properties.bootstrap.servers' = '192.168.102.120:9092,192.168.102.121:9092,192.168.102.122:9092,192.168.102.123:9092',\n" +
//                "  'format' = 'json'\n" +
//                ")");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
                        "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
//        tableEnv.executeSql("insert into sinkTable Select JSON_ARRAY(order_number, price) as jjj FROM sourceTable");
        tableEnv.executeSql("insert into sinkTable " +
//                "select order_number, JSON_ARRAYAGG(JSON_OBJECT('order_number' VALUE order_number)) as jjj FROM sourceTable group by order_number");
                "select * FROM sourceTable");
//                "select order_number, buyer FROM sourceTable");

    }
}
